

package com.qiyu.mq.core.producer;

import com.oracle.tools.packager.Log;
import com.qiyu.mq.core.annotation.MqProvider;
import com.qiyu.mq.core.config.MQException;
import com.qiyu.mq.core.config.MqProducerConfig;
import com.qiyu.mq.core.serializer.Serializer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MQProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.aop.framework.Advised;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import java.lang.reflect.Field;

/**
 * @author idea
 * @version V1.0
 * @date 2020/8/10
 */
public class DefaultProducerFactory implements ProducerFactory, ApplicationContextAware, DisposableBean, BeanPostProcessor {

    private volatile MQProducer mqProducer;

    private ApplicationContext applicationContext;

    private MqProducerConfig mqProducerConfig;

    private String nameServers;

    public DefaultProducerFactory(String nameServers, MqProducerConfig mqProducerConfig) {
        this.nameServers = nameServers;
        this.mqProducerConfig = mqProducerConfig;
    }

    @Override
    public MQProducer createProducer() {
        if (mqProducer == null) {
            synchronized (this) {
                if (mqProducer == null) {
                    try {
                        DefaultMQProducer defaultMQProducer = new DefaultMQProducer(mqProducerConfig.getGroupName());
                        defaultMQProducer.setNamesrvAddr(nameServers);
                        defaultMQProducer.setSendMsgTimeout(8000);
                        //如果这个参数不设置为false则需要开启10909端口
                        defaultMQProducer.setVipChannelEnabled(false);
                        mqProducer = defaultMQProducer;
                        mqProducer.start();
                        System.out.println("defaultMQProducer create success,defaultMQProducer is " + defaultMQProducer.toString());
                    } catch (Exception e) {
                        throw new MQException(e);
                    }
                }
            }
        }
        return mqProducer;
    }

    @Override
    public void destroy() throws Exception {

    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }


    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        //直接可以扫描出整个工程创建过程中用到的class
        Class clazz = AopUtils.isAopProxy(bean) ? AopUtils.getTargetClass(bean) : bean.getClass();
        Field[] fields = clazz.getDeclaredFields();
        for (Field field : fields) {
            field.setAccessible(true);
            if (field.isAnnotationPresent(MqProvider.class)) {
                MqProvider mqProvider = field.getAnnotation(MqProvider.class);
                String topic = mqProvider.topic();
                Class<?> serializerClass = mqProvider.serializer();
                Object serializer = null;
                try {
                    serializer = mqProvider.serializer().newInstance();
                } catch (InstantiationException e) {
                    e.printStackTrace();
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                }
                MqProducerTemplate mqProducerTemplate = new MqProducerTemplate();
                mqProducerTemplate.setProducerFactory(this);
                mqProducerTemplate.setTopic(topic);
                mqProducerTemplate.setTag(mqProvider.tag());
                mqProducerTemplate.setSerializer((Serializer) serializer);
                try {
                    if (AopUtils.isAopProxy(bean) && bean instanceof Advised) {
                        Advised advised = (Advised) bean;
                        Object originBean = null;
                        originBean = advised.getTargetSource().getTarget();
                        field.set(originBean, mqProducerTemplate);

                    }
                    field.set(bean, mqProducerTemplate);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
        return null;
    }
}
